package com.ybt.communicate.tcp.client;

import java.io.InputStream;
import java.io.UnsupportedEncodingException;
import java.net.Socket;
import java.nio.channels.Channels;
import java.nio.channels.ReadableByteChannel;
import java.nio.channels.WritableByteChannel;
import java.util.UUID;

import org.apache.log4j.Logger;

import com.ruleengine.context.IWorkflowConext;
import com.ybt.communicate.core.IConnection;
import com.ybt.communicate.core.callback.IConnectionCallback;
import com.ybt.communicate.core.exception.ProtocolException;
import com.ybt.communicate.tcp.configration.TCPConstants;
import com.ybt.communicate.tcp.configration.TCPRuntimeConfig;

public class TCPClientLongConnection implements IConnection {
    private static Logger logger = Logger
    .getLogger(TCPClientLongConnection.class);
    private Socket socket;
    private TCPRuntimeConfig config;
    private UUID startId;
    private TCPClientLong protocol;

    public TCPClientLongConnection(Socket socket, TCPRuntimeConfig config,
            UUID startId) {
        super();
        this.socket = socket;
        this.config = config;
        this.startId = startId;
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.ybt.communicate.core.IConnection#receive()
     */
    public Object receive() throws ProtocolException {
        // TODO Auto-generated method stub
        if (!config.isDuplex()
                && TCPConstants.IODIRECTION_SEND
                .equals(config.getIODirection())) {
            // 单通道，发送数据
            throw new ProtocolException("unsupport method for "
                    + config.getProtocolType() + ", protocol id ["
                    + config.getProtocolId() + "]");
        }

        logger.debug("begin to receive data from "
                + socket.getRemoteSocketAddress());

        byte[] contents = null;
        try {
            InputStream in = socket.getInputStream();
            ReadableByteChannel channel = Channels.newChannel(in);
            contents = config.getReadAccessor().receive(channel);
        } catch (Exception e) {
            // 通知协议发生异常
            protocol.notify(startId, e);
            throw new ProtocolException("receive data from client "
                    + config.getIp()+":"+config.getPort() + " error ", e);
        }

        if (TCPConstants.TO_STRING.equals(config.getStaticConfig()
                .getCommonConfig().getAction())) {
            // 将byte转换成string
            String encode = config.getStaticConfig().getCommonConfig()
            .getEncode();
            if (encode == null) {
                // 使用平台默认编码
                return new String(contents);
            } else {
                // 使用自定义编码
                try {
                    return new String(contents, encode);
                } catch (UnsupportedEncodingException e) {
                    // TODO Auto-generated catch block
                    throw new ProtocolException(
                            "convert the data received from client "
                            + config.getIp()+":"+config.getPort() + " error ", e);
                }
            }
        }
        return contents;
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.ybt.communicate.core.IConnection#send(com.ruleengine.context.IWorkflowConext,
     *      java.lang.Object)
     */
    public Object send(IWorkflowConext context, Object data)
    throws ProtocolException {
        // TODO Auto-generated method stub
        try {
            // 发送数据
            logger.debug("begin to send data to "
                    + socket.getRemoteSocketAddress());
            WritableByteChannel channelOut = Channels.newChannel(socket
                    .getOutputStream());
            config.getWriteAccessor().send(channelOut, (byte[]) data);
        } catch (Exception e) {
            // 通知协议发生异常
            protocol.notify(startId, e);

            throw new ProtocolException("send data to client "
                    + config.getIp()+":"+config.getPort() + " error ", e);
        }

        // 是否接收数据
        if (config.isDuplex()) {
            // 双通道，client端
            // 同步tcp请求，需要等待返回数据
            logger.debug("begin to receive data from "
                    + socket.getRemoteSocketAddress());
            byte[] contents = null;
            try {
                InputStream in = socket.getInputStream();
                ReadableByteChannel channel = Channels.newChannel(in);
                contents = config.getReadAccessor().receive(channel);

            } catch (Exception e) {
                // 通知协议发生异常
                protocol.notify(startId, e);

                throw new ProtocolException("receive data from client "
                        + config.getIp()+":"+config.getPort() + " error ", e);
            }

            if (TCPConstants.TO_STRING.equals(config.getStaticConfig()
                    .getCommonConfig().getAction())) {
                // 将byte转换成string
                String encode = config.getStaticConfig().getCommonConfig()
                .getEncode();
                if (encode == null) {
                    // 使用平台默认编码
                    return new String(contents);
                } else {
                    // 使用自定义编码
                    try {
                        return new String(contents, encode);
                    } catch (UnsupportedEncodingException e) {
                        // TODO Auto-generated catch block
                        throw new ProtocolException(
                                "convert the data received from client "
                                + config.getIp()+":"+config.getPort() + " error ", e);
                    }
                }
            }
            return contents;
        }
        return null;
    }

    /*
     * (non-Javadoc)
     * 
     * @see com.ybt.communicate.core.IConnection#sendNonBlocking(com.ruleengine.context.IWorkflowConext,
     *      java.lang.Object,
     *      com.ybt.communicate.core.callback.IConnectionCallback)
     */
    public void sendNonBlocking(IWorkflowConext context, Object data,
            IConnectionCallback callBack) throws ProtocolException {
        // TODO Auto-generated method stub
        // FIXME 目前不支持非阻塞调用
        throw new ProtocolException("unsupport method for "
                + config.getProtocolType() + ", protocol id ["
                + config.getProtocolId() + "]");
    }

    public void destroy() throws ProtocolException {
        // TODO Auto-generated method stub
        try {
            if (socket != null) {
                socket.close();
            }
        } catch (Exception e) {
            throw new ProtocolException("close socket to server "
                    + config.getIp()+":"+config.getPort() + " error ", e);
        }
    }

    
    /*
     * (non-Javadoc)
     * 
     * @see com.ybt.communicate.core.IConnection#close()
     */
    public void close() throws ProtocolException {
        // TODO Auto-generated method stub
        try {
            protocol.relaseConnnect(this);
        } catch (Exception e) {
            throw new ProtocolException("close socket to server "
                    + config.getIp()+":"+config.getPort() + " error ", e);
        }
    }

    public String getProtocolId() {
        // TODO Auto-generated method stub
        return config.getProtocolType();
    }

    public TCPClientLong getProtocol() {
        return protocol;
    }

    public void setProtocol(TCPClientLong protocol) {
        this.protocol = protocol;
    }

    public UUID getStartId() {
        return startId;
    }

    public void setStartId(UUID startId) {
        this.startId = startId;
    }

}
